package io.wen.bd.s8m1;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class Task1_FlinkIntro {
    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        final DataStream<String> socketStream = env.socketTextStream("hadoop03", 9999);

        final DataStream<Tuple2<String, Long>> wordsStream = socketStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {

            @Override
            public void flatMap(String value, Collector<Tuple2<String, Long>> collector) throws Exception {
                final String[] splits = value.split(" ");
                for (String word : splits) {
                    collector.collect(Tuple2.of(word, 1L));
                }
            }
        });

        final DataStream<Tuple2<String, Long>> wordCounts = wordsStream
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(3)))
                .sum(1);
        wordCounts.print();

        env.execute();
    }
}
